////////////////////////////////////////////////////////////////////////////
//
//  CryEngine Source File.
//  Copyright (C), Crytek.
// -------------------------------------------------------------------------
//  File name:   StreamIOThread.cpp
//  Created:     22/07/2010 by Timur.
//  Description: Streaming Thread for IO
// -------------------------------------------------------------------------
//  History:
//
////////////////////////////////////////////////////////////////////////////

#include <StdAfx.h>
#include "StreamIOThread.h"
#include "StreamEngine.h"
#include "../System.h"
#include "IPlatformOS.h"

extern SSystemCVars g_cvars;

//////////////////////////////////////////////////////////////////////////
CStreamingIOThread::CStreamingIOThread( CStreamEngine *pStreamEngine,EStreamSourceMediaType mediaType,const char *name, int32 prio)
{
	m_pStreamEngine = pStreamEngine;
	m_bCancelThreadRequest = false;
	m_bNeedSorting = false;
	m_bNeedReset = false;
	m_name = name;
	m_eMediaType = mediaType;

	m_iUrgentRequests = 0;

	m_bPaused = false;
	m_bAbortReads = false;

	m_nReadCounter = 0;
	m_nStreamingCPU = -1;



	Start( (unsigned)(1 << g_cvars.sys_streaming_cpu),name );

}

CStreamingIOThread::~CStreamingIOThread()
{
	Cancel();
	WaitForThread();
	Stop();
}

//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::AddRequest( CAsyncIOFileRequest* pRequest,bool bStartImmidietly )
{
	EStreamSourceMediaType eMediaType = pRequest->GetMediaType();
	const TFallbackIOVecConstIt itEnd = m_FallbackIOThreads.end();
	for(TFallbackIOVecConstIt it = m_FallbackIOThreads.begin(); it!=itEnd; ++it)
	{
		if(it->second == eMediaType)
		{
			(it->first)->AddRequest( pRequest, true );
			return;
		}
	}

	//handle by hdd file io thread
	pRequest->AddRef(); // Acquire ownership on file request.
	m_newFileRequests.push_back(pRequest);
	pRequest->m_status = CAsyncIOFileRequest::eStatusInFileQueue;
	pRequest->m_nReadCounter = 0;
	if (pRequest->m_eMediaType != eStreamSourceTypeMemory)
		pRequest->m_eMediaType = m_eMediaType;
	// does this ignore the tmp out of memory
	if (pRequest->IgnoreOutofTmpMem())
	{
		CryInterlockedIncrement(&m_iUrgentRequests);
	}
#ifndef PS3
	//we cannot afford any high latency by gathering requests on PS3
	//hdd should serve asap, fmod starves otherwise
	if (bStartImmidietly)
#endif
	{
		READ_WRITE_BARRIER
		m_bNewRequests = true;
		m_awakeEvent.Set();
	}
}


//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::SignalStartWork(bool bForce)
{
	if (!m_newFileRequests.empty() || bForce)
	{
		READ_WRITE_BARRIER
		m_bNewRequests = true;

		m_awakeEvent.Set();
	}
}

//////////////////////////////////////////////////////////////////////////

void CStreamingIOThread::Pause(bool bPause)
{
	m_bPaused = bPause;
}

//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::Run()
{
	SetName( m_name );

	CTimeValue t0 = gEnv->pTimer->GetAsyncTime();

	m_nLastReadDiskOffset = 0;

	//
	// Main thread loop
	while (!m_bCancelThreadRequest)
	{
		if (m_nStreamingCPU != g_cvars.sys_streaming_cpu)
		{
			m_nStreamingCPU = g_cvars.sys_streaming_cpu;



		}

		if (m_bNewRequests || !m_newFileRequests.empty())
		{
			READ_WRITE_BARRIER
			ProcessNewRequests();
		}
		else
		{
			if (m_eMediaType == eStreamSourceTypeDisc && gEnv->pSystem->GetPlatformOS())
			{
				gEnv->pSystem->GetPlatformOS()->SetOpticalDriveIdle(true);
			}

#if defined(_RELEASE) || defined(PS3) || defined(CAFE)
			m_awakeEvent.Wait();
#else
			// compute max time to wait - revive thread every second at least once to update stats
			bool bWaiting = true;
			while (bWaiting)
			{
				CTimeValue t1 = gEnv->pTimer->GetAsyncTime();
				CTimeValue deltaT = t1 - t0;
				uint64 msec = deltaT.GetMilliSecondsAsInt64();
				if (msec < 1000) 
				{
					bWaiting = !m_awakeEvent.Wait(1000 - (uint32)msec);
				}

				if (bWaiting)
				{
					// update the delta time again
					t1 = gEnv->pTimer->GetAsyncTime();
					deltaT = t1 - t0;

					m_InMemoryStats.Update(deltaT);
					m_NotInMemoryStats.Update(deltaT);

					t0 = t1;	
				}
			}	
#endif

			if (m_eMediaType == eStreamSourceTypeDisc && gEnv->pSystem->GetPlatformOS())
			{
				gEnv->pSystem->GetPlatformOS()->SetOpticalDriveIdle(false);
			}

			m_awakeEvent.Reset();
		}

		if (m_bNeedReset)
		{
			ProcessReset();
		}

		while (!m_bCancelThreadRequest && !m_fileRequestQueue.empty())
		{
			// Handle any cancelled requests
			bool bCancel(false);
			for( std::vector<CAsyncIOFileRequest*>::iterator it = m_fileRequestQueue.begin();
				it != m_fileRequestQueue.end(); it++ )
			{
				CAsyncIOFileRequest_AutoPtr pFileRequest = *it;
				if( pFileRequest->IsCanceled() )
				{
					// check if request was high prio, then decr open count
					if (pFileRequest->IgnoreOutofTmpMem())
						CryInterlockedDecrement(&m_iUrgentRequests);

					pFileRequest->Release();
					m_fileRequestQueue.erase( it );	
					bCancel = true;
					break;
				}
			}

			if( bCancel )
				continue;

			CAsyncIOFileRequest_AutoPtr pFileRequest = m_fileRequestQueue.back();
			m_fileRequestQueue.pop_back();

			if (pFileRequest->IsCanceled())
			{
				// check if request was high prio, then decr open count
				if (pFileRequest->IgnoreOutofTmpMem())
					CryInterlockedDecrement(&m_iUrgentRequests);

				pFileRequest->Release(); // Release ownership on file request.
				continue;
			}

			//////////////////////////////////////////////////////////////////////////
			// When temporary memory goes out of budget we must loop here and wait until previous file requests are finished and free up memory.
			// Only allow processing of requests which are flagged for processing when out of tmp memory
			//////////////////////////////////////////////////////////////////////////
			if (g_cvars.sys_streaming_memory_budget != 0 && 
						m_pStreamEngine->GetCurrentTempMemorySize()/(1024*1024) > (uint32)g_cvars.sys_streaming_memory_budget && 
						!m_bCancelThreadRequest)
			{
				m_pStreamEngine->FlagTempMemOutOfBudget();
				if (m_iUrgentRequests > 0)
				{
					if (m_bNewRequests || !m_newFileRequests.empty())
					{
						READ_WRITE_BARRIER
						ProcessNewRequests();
					}

					// readd the current request
					if (pFileRequest)
						m_fileRequestQueue.push_back(pFileRequest);
					pFileRequest = 0;
					// search for the first request which ignores the current out of mem state
					// Search for next highest priority request
					std::vector<CAsyncIOFileRequest*>::reverse_iterator rit;
					for (rit = m_fileRequestQueue.rbegin(); rit != m_fileRequestQueue.rend(); ++rit)
					{
						if ((*rit)->IgnoreOutofTmpMem())
						{
							pFileRequest = *rit;
							std::vector<CAsyncIOFileRequest*>::iterator it( rit.base() );
							--it;
							m_fileRequestQueue.erase(it);
							break;
						}
					}
				} 
				else
				{
					// read the current request
					if( pFileRequest )
						m_fileRequestQueue.push_back( pFileRequest );
					pFileRequest = 0;
				}
			}

			// Simply let the io thread sleep when paused before doing any actual IO
			while (m_bPaused)
				CrySleep(10);

			// If at this point, the filerequest is zero, the above prioritization of
			// urgent requests couldn't find a new task to displace the current
			// one. As the current one had been pushed back previously, we can safely
			// assume that restarting the loop will grab it again (eventually). 
			if (!pFileRequest)
			{



				continue; 

			}

			// Handle file request.	
			if (m_bAbortReads)
				pFileRequest->ReadFailed(ERROR_ABORTED_ON_SHUTDOWN);
			else
				pFileRequest->ReadFile();
			// check if request was high prio, then decr open count
			if (pFileRequest->IgnoreOutofTmpMem())
				CryInterlockedDecrement(&m_iUrgentRequests);
			if (pFileRequest->m_eMediaType != eStreamSourceTypeMemory && pFileRequest->m_nErrorCode == 0)
			{
				pFileRequest->m_nReadHeadOffsetKB = (int32)(((int64)pFileRequest->m_nDiskOffset - m_nLastReadDiskOffset) >> 10); // in KB
				m_nLastReadDiskOffset = pFileRequest->m_nDiskOffset + pFileRequest->m_nSizeOnMedia;

				m_NotInMemoryStats.m_nTempReadOffset += std::abs(pFileRequest->m_nReadHeadOffsetKB);
				m_NotInMemoryStats.m_nTotalReadOffset += std::abs(pFileRequest->m_nReadHeadOffsetKB);

				m_NotInMemoryStats.m_nTempRequestCount++;

				// Calc IO bandwidth only for non memory files.
				if (pFileRequest->m_status == CAsyncIOFileRequest::eStatusReadComplete)
				{
					m_NotInMemoryStats.m_nTempBytesRead += pFileRequest->m_nSizeOnMedia;
					m_NotInMemoryStats.m_TempReadTime += pFileRequest->m_readTime;
				}
			} else {
				m_InMemoryStats.m_nTempRequestCount++;

				// Calc IO bandwidth only for in memory files.
				if (pFileRequest->m_status == CAsyncIOFileRequest::eStatusReadComplete)
				{
					m_InMemoryStats.m_nTempBytesRead += pFileRequest->m_nSizeOnMedia;
					m_InMemoryStats.m_TempReadTime += pFileRequest->m_readTime;
				}
			}

			pFileRequest->m_nReadCounter = m_nReadCounter++;
		
			m_pStreamEngine->ReportAsyncFileRequestReadDone( pFileRequest );
			pFileRequest->Release(); // Release ownership on file request.

			//////////////////////////////////////////////////////////////////////////
			if (m_bNewRequests)
			{
				READ_WRITE_BARRIER
				ProcessNewRequests();
			}
			if (m_bNeedReset)
			{
				ProcessReset();
			}
			if (m_bNeedSorting)
			{
				SortRequests();
			}

			//////////////////////////////////////////////////////////////////////////
			if (g_cvars.sys_streaming_max_bandwidth != 0)
			{
				CTimeValue t1 = gEnv->pTimer->GetAsyncTime();
				CTimeValue deltaT = t1 - t0;

				// Sleep in case we are streaming too fast.
				const float fTheoreticalReadTime = float(pFileRequest->m_nSizeOnMedia) / g_cvars.sys_streaming_max_bandwidth * 0.00000095367431640625f;	// / (1024*1024)

				if(fTheoreticalReadTime - deltaT.GetSeconds() > FLT_EPSILON)
				{
					uint32 nSleepTime = uint32(1000.f * (fTheoreticalReadTime - deltaT.GetSeconds()));
					CrySleep(nSleepTime);
				}
			}

			CTimeValue t1 = gEnv->pTimer->GetAsyncTime();
			CTimeValue deltaT = t1 - t0;

			// update the stats every second
			if (deltaT.GetMilliSecondsAsInt64() > 1000)
			{
				m_InMemoryStats.Update(deltaT);
				m_NotInMemoryStats.Update(deltaT);

				t0 = t1;
			}
		}
	}
}

void CStreamingIOThread::SStats::Update(const CTimeValue& deltaT)
{
	m_nReadBytesInLastSecond = (uint32)m_nTempBytesRead;
	m_nRequestCountInLastSecond = m_nTempRequestCount;
	m_nTotalReadBytes += (uint32)m_nTempBytesRead;
	m_nTotalRequestCount += m_nTempRequestCount;
	m_TotalReadTime += m_TempReadTime;

	if (m_TempReadTime.GetValue() != 0)
		m_nActualReadBandwith = (uint32)(m_nTempBytesRead / m_TempReadTime.GetSeconds());
	else
		m_nActualReadBandwith = 0;
	m_nCurrentReadBandwith = (uint32)(m_nTempBytesRead / deltaT.GetSeconds());
	m_fReadingDuringLastSecond = m_TempReadTime.GetSeconds() / deltaT.GetSeconds() * 100;

	if (m_nTempRequestCount > 0)
		m_nReadOffsetInLastSecond = m_nTempReadOffset / m_nTempRequestCount;
	else
		m_nReadOffsetInLastSecond = 0;

	m_TempReadTime.SetValue(0);
	m_nTempBytesRead = 0;
	m_nTempReadOffset = 0;
	m_nTempRequestCount = 0;
}

//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::Cancel()
{
	m_bCancelThreadRequest = true;
	m_awakeEvent.Set();
}

//////////////////////////////////////////////////////////////////////////
struct SCompareAsyncFileRequest
{
	bool operator()( const CAsyncIOFileRequest_AutoPtr &pFile1,const CAsyncIOFileRequest_AutoPtr &pFile2 ) const
	{
		return pFile1->m_nSortKey > pFile2->m_nSortKey;
	}
};

//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::SortRequests()
{
	FUNCTION_PROFILER( gEnv->pSystem, PROFILE_SYSTEM );

	std::sort( m_fileRequestQueue.begin(),m_fileRequestQueue.end(),SCompareAsyncFileRequest() );
/*
	int nStartOfQueue = 0;
	int64 nDiskOffsetLimit = m_nLastReadDiskOffset - 32*1024; // 32KB less only
		
	int nCount = (int)m_fileRequestQueue.size();
	for (int i = nCount-1; i >= 0; i--)
	{
		if (m_fileRequestQueue[i]->m_nDiskOffset > nDiskOffsetLimit)
		{
			nStartOfQueue = i+1;
			break;
		}
	}
	if (nStartOfQueue < nCount && nStartOfQueue > 0)
	{
		int nElements = nCount - nStartOfQueue;
		// Move all elements up to nStartOfQueue, from begining of the request array to the end.
		m_temporaryArray.resize(0);
		// Copy to temp array elements up to nStartOfQueue
		m_temporaryArray.insert( m_temporaryArray.end(),m_fileRequestQueue.begin()+nStartOfQueue,m_fileRequestQueue.end() );
		// Remove elements up to nStartOfQueue from request list
		m_fileRequestQueue.erase( m_fileRequestQueue.begin()+nStartOfQueue,m_fileRequestQueue.end() );
		// Add elemenets at the end from temp array.
		m_fileRequestQueue.insert( m_fileRequestQueue.begin(),m_temporaryArray.begin(),m_temporaryArray.end() );
	}
	*/
	m_bNeedSorting = false;
}

void CStreamingIOThread::NeedSorting()
{
	m_bNeedSorting = true;
}

//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::ProcessNewRequests()
{
	m_bNewRequests = false;

	std::vector<CAsyncIOFileRequest*> temporaryArray;
	temporaryArray.resize(0);
	m_newFileRequests.swap(temporaryArray);

	std::vector<CAsyncIOFileRequest*> &newFiles = temporaryArray;

	if (!newFiles.empty())
	{
		CAsyncIOFileRequest *pRequest = 0;
		if (!m_fileRequestQueue.empty())
			pRequest = m_fileRequestQueue.back();

		uint64 nCurrentKeyInProgress = m_fileRequestQueue.size() ? m_fileRequestQueue.back()->m_nSortKey : 0;

		// Compute sorting key for new file entries.
		int iWakeFallback(0);
		const TFallbackIOVecConstIt itEnd = m_FallbackIOThreads.end();
		const size_t fallbackNum = m_FallbackIOThreads.size();
		PREFAST_SUPPRESS_WARNING(6255)
		uint8 *pFallbackSignals = fallbackNum?(uint8*)alloca(fallbackNum):NULL;
		for(uint32 fb = 0; fb < fallbackNum; ++fb)
			pFallbackSignals[fb] = 0;
		for (size_t i = 0,num = newFiles.size(); i < num; i++)
		{
			CAsyncIOFileRequest *pFilepRequest = newFiles[i];					
			pFilepRequest->ComputeSortKey(nCurrentKeyInProgress);
			bool bFallback = false;
			int idx = -1;
			for(TFallbackIOVecConstIt it = m_FallbackIOThreads.begin(); it!=itEnd && !bFallback; ++it)
			{
				++idx;
				if(it->second == pFilepRequest->GetMediaType())
				{
					if (pFilepRequest->IgnoreOutofTmpMem())
					{
						CryInterlockedDecrement(&m_iUrgentRequests);
					}
					(it->first)->AddRequest(pFilepRequest, true);
					pFilepRequest->Release(); // Release local ownership of request (moved to fallback IO thread)
					iWakeFallback++;
					bFallback = true;
					pFallbackSignals[idx] = 1;
				}
			}
			if(!bFallback)
				m_fileRequestQueue.push_back(pFilepRequest);
		}

		for(uint32 fb = 0; fb < fallbackNum; ++fb)
		{
			if(pFallbackSignals[fb] != 0)
				(m_FallbackIOThreads[fb].first)->SignalStartWork(false);
		}

		SortRequests();
		/*
		if (m_fileRequestQueue.back() != pRequest && pRequest != 0)
		{
			// Highest priority changed.
			if (m_fileRequestQueue.back()->m_nDiskOffset < (m_nLastReadDiskOffset-32*1024))
			{
				//CryLog( "Bad Offset in Queue" );
			}
		}
		*/
	}
}

void CStreamingIOThread::ProcessReset()
{
	if (!m_fileRequestQueue.empty())
	{
		for (std::vector<CAsyncIOFileRequest*>::iterator it = m_fileRequestQueue.begin(), itEnd = m_fileRequestQueue.end(); it != itEnd; ++ it)
			(*it)->Release();
	}

	stl::free_container(m_fileRequestQueue);

	if (!m_temporaryArray.empty())
	{
		for (std::vector<CAsyncIOFileRequest*>::iterator it = m_temporaryArray.begin(), itEnd = m_temporaryArray.end(); it != itEnd; ++ it)
			(*it)->Release();
	}

	stl::free_container(m_temporaryArray);

	m_bNeedReset = false;
	m_resetDoneEvent.Set();
}

//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::CancelAll()
{
	{
		CryMT::vector<CAsyncIOFileRequest*>::AutoLock lock(m_newFileRequests.get_lock());

		if (!m_newFileRequests.empty())
		{
			CAsyncIOFileRequest * const * it = &m_newFileRequests.front();
			CAsyncIOFileRequest * const * itEnd = it + m_newFileRequests.size();
			for (; it != itEnd; ++ it)
				(*it)->Release();
		}
	}

	m_newFileRequests.free_memory();
	m_iUrgentRequests = 0;
}

void CStreamingIOThread::AbortAll(bool bAbort)
{
	m_bAbortReads = bAbort;
}

void CStreamingIOThread::BeginReset()
{
	CancelAll();

	m_resetDoneEvent.Reset();
	m_bNeedReset = true;
	m_awakeEvent.Set();
}

void CStreamingIOThread::EndReset()
{
	m_resetDoneEvent.Wait();
}
//////////////////////////////////////////////////////////////////////////
void CStreamingIOThread::RegisterFallbackIOThread(EStreamSourceMediaType mediaType, CStreamingIOThread* pIOThread)
{
	//check if media has not yet been registered
	if(!pIOThread)
		return;//no need for NULL register anymore
	const TFallbackIOVecConstIt itEnd = m_FallbackIOThreads.end();
	for(TFallbackIOVecConstIt it = m_FallbackIOThreads.begin(); it!=itEnd; ++it)
	{
		if(it->second == mediaType)
			return;
	}
	m_FallbackIOThreads.push_back(std::make_pair(pIOThread, mediaType));
}


//////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////

//////////////////////////////////////////////////////////////////////////
CStreamingWorkerThread::CStreamingWorkerThread( CStreamEngine *pStreamEngine,const char *name,EWorkerType type )
{
	m_type = type;
	m_name = name;
	m_pStreamEngine = pStreamEngine;
	m_bCancelThreadRequest = false;
	m_bNeedsReset = false;

	// Starts thread (use 40kb stack size to prevent a stack overflow on ps3(default stack size is 32kb))
	Start( (unsigned)1 << g_cvars.sys_streaming_cpu,name,THREAD_PRIORITY_NORMAL,40*1024);
}

CStreamingWorkerThread::~CStreamingWorkerThread()
{
	Cancel();
	WaitForThread();
	Stop();
}

//////////////////////////////////////////////////////////////////////////
void CStreamingWorkerThread::AddRequest( CAsyncIOFileRequest* pRequest )
{
	m_fileRequestQueue.push_back(pRequest);
	m_awakeEvent.Set();
}

//////////////////////////////////////////////////////////////////////////
void CStreamingWorkerThread::Run()
{
	SetName( m_name );









	// Main thread loop
	while (!m_bCancelThreadRequest)
	{
		m_awakeEvent.Wait();
		m_awakeEvent.Reset();

		CAsyncIOFileRequest_AutoPtr pFileRequest;
		while (!m_bCancelThreadRequest && !m_bNeedsReset && m_fileRequestQueue.try_pop_front(pFileRequest))
		{
			switch (m_type)
			{
			case eWorkerDecompress:
				{
					pFileRequest->Decompress();
					m_pStreamEngine->ReportAsyncFileRequestDecompressed( pFileRequest );
				}
				break;
			case eWorkerAsyncCallback:
				{
					float fTime = gEnv->pTimer->GetAsyncCurTime();
					m_pStreamEngine->ReportAsyncFileRequestComplete( pFileRequest );
					float fTime1 = gEnv->pTimer->GetAsyncCurTime();

					// Temp code for PS3, remove later
					if ((fTime1 - fTime) > 1.f && !pFileRequest->m_strFileName.empty())
					{
						string str; str.Format("[ACALL] %s time=%.5f\n",pFileRequest->m_strFileName.c_str(),(fTime1-fTime) ); OutputDebugString(str.c_str());
					}
				}
				break;
			}
		}

		if (m_bNeedsReset)
		{
			m_fileRequestQueue.free_memory();
			m_bNeedsReset = false;
			m_resetDoneEvent.Set();
		}
	}
}

//////////////////////////////////////////////////////////////////////////
void CStreamingWorkerThread::Cancel()
{
	m_bCancelThreadRequest = true;
	m_awakeEvent.Set();
}

//////////////////////////////////////////////////////////////////////////
void CStreamingWorkerThread::CancelAll()
{
	m_fileRequestQueue.free_memory();
}

void CStreamingWorkerThread::BeginReset()
{
	CancelAll();

	m_resetDoneEvent.Reset();
	m_bNeedsReset = true;
	m_awakeEvent.Set();
}

void CStreamingWorkerThread::EndReset()
{
	m_resetDoneEvent.Wait();
}
